package com.atguigu.market_analysis

import com.atguigu.market_analysis.AdClickAnalysisZB.outputTag
import org.apache.flink.api.common.functions.AggregateFunction
import org.apache.flink.api.common.state.ValueStateDescriptor
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment, _}
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

import java.sql.Timestamp

// 广告类
case class AdClickLog(userId: Long, adId: Long, province: String, city: String, timestamp: Long)

// 根据省份统计广告点击次数
case class AdClickCountByProvince(windowEnd: String, province: String, count: Long)

// 侧输出流黑名单报警信息样例类
case class BlackListUserWarning(userId: Long, adId: Long, msg: String)


object AdClickAnalysisZB {
  val outputTag = new OutputTag[BlackListUserWarning]("warning")

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    // 从文件中读取数据
    val inputStream = env.readTextFile(getClass.getResource("/AdClickLog.csv").getPath)

    // 转换成样例类, 并提取时间戳和waterMarker
    val adLogStream = inputStream
      .map(line => {
        val arr = line.split(",")
        AdClickLog(arr(0).toLong, arr(1).toLong, arr(2), arr(3), arr(4).toLong)
      })
      .assignAscendingTimestamps(_.timestamp * 1000L)

    // 过滤刷单行为, 并将刷单用户加入到黑名单(侧输出流)
    val filteredStream = adLogStream
      .keyBy(ad => (ad.userId, ad.adId))
      .process(new FilterBlackListUserKPF(100))

    val adCountResultStream = filteredStream
      .keyBy(_.province)
      .timeWindow(Time.hours(1), Time.seconds(5))
      .aggregate(new AdCountAggFun, new AdCountWindowFun)

    adLogStream.print("ad")
    filteredStream.getSideOutput(outputTag).print("blackList")
    adCountResultStream.print("result")

    env.execute("ad count job")
  }
}


class FilterBlackListUserKPF(maxCount: Int) extends KeyedProcessFunction[(Long, Long), AdClickLog, AdClickLog] {
  // 定义状态,保存广告点击量, 每天0点清空状态的时间戳,标记当前用户是否已经进入黑名单
  lazy private val curCount = getRuntimeContext.getState(new ValueStateDescriptor[Long]("curCount", classOf[Long]))
  lazy private val resetTimerTs = getRuntimeContext.getState(new ValueStateDescriptor[Long]("resetTimerTs", classOf[Long]))
  lazy private val isInBlackList = getRuntimeContext.getState(new ValueStateDescriptor[Boolean]("isInBlackList", classOf[Boolean]))

  override def processElement(value: AdClickLog, context: KeyedProcessFunction[(Long, Long), AdClickLog, AdClickLog]#Context, collector: Collector[AdClickLog]): Unit = {
    if (isInBlackList.value()) return

    // 第一个数据来了, 注册定时器, 用于清空所有状态
    if (curCount.value == 0) {
      val ts = (context.timerService().currentProcessingTime() / (1000 * 60 * 60 * 24) + 1) * (24 * 60 * 60 * 1000) - 8 * 60 * 60 * 1000
      resetTimerTs.update(ts)
      context.timerService().registerEventTimeTimer(ts)
    }

    // 增加计数
    curCount.update(curCount.value() + 1)

    // 如果点击数达到最大值, 就加入黑名单, 并输出侧输出流
    if (curCount.value() >= maxCount) {
      isInBlackList.update(true)
      context.output(outputTag, BlackListUserWarning(value.userId, value.adId, s"点击超过了${maxCount}次"))
    } else {
      collector.collect(value)
    }

  }

  override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[(Long, Long), AdClickLog, AdClickLog]#OnTimerContext, out: Collector[AdClickLog]): Unit = {
    // 清空状态
    if (resetTimerTs.value() == timestamp) {
      curCount.update(0)
      isInBlackList.update(false)
    }
  }
}

class AdCountAggFun extends AggregateFunction[AdClickLog, Long, Long] {
  override def createAccumulator(): Long = 0L

  override def add(in: AdClickLog, acc: Long): Long = acc + 1

  override def getResult(acc: Long): Long = acc

  override def merge(acc: Long, acc1: Long): Long = acc + acc1
}

class AdCountWindowFun extends WindowFunction[Long, AdClickCountByProvince, String, TimeWindow] {
  override def apply(key: String, window: TimeWindow, input: Iterable[Long], out: Collector[AdClickCountByProvince]): Unit = {
    val end = new Timestamp(window.getEnd).toString
    out.collect(AdClickCountByProvince(end, key, input.head))
  }
}